/***
* ==++==
*
* Copyright (c) Microsoft Corporation.  All rights reserved.
*
* ==--==
* =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
*
* astreambuf.h
*
* Asynchronous I/O: stream buffer
*
* =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
****/
#pragma once

#include <ios>
#include <memory>
#include <math.h>
#include <sstream>
#include <pplxtasks.h>
#if !defined(__cplusplus_winrt)
#include "xstreambuf.h"
#endif
#include "asyncrt_utils.h"

// Suppress unreferenced formal parameter warning as they are required for documentation
#pragma warning(push)
#pragma warning(disable : 4100)

namespace pplx
{
    // A convenient extension to PPLX

    template <typename Function>
    task<bool> do_while(Function func)
    {
        task<bool> first = func();
        return first.then([=](bool guard) -> task<bool> {
            if (guard)
                return do_while(func);
            else
                return first;
            });
    }
}

namespace async
{
    /// <summary>
    /// Stream buffer base class.
    /// </summary>
    class basic_streambuf
    {
    public:
        /// <summary>
        /// Virtual constructor for stream buffers.
        /// </summary>
        virtual ~basic_streambuf() { }

        /// <summary>
        /// can_read is used to determine whether a stream buffer will support read operations (get).
        /// </summary>
        bool can_read() const { return _can_read(); }

        /// <summary>
        /// can_write is used to determine whether a stream buffer will support write operations (put).
        /// </summary>       
        bool can_write() const { return _can_write(); }
        
        /// <summary>
        /// can_seek is used to determine whether a stream buffer supports seeking.
        /// </summary>
        bool can_seek() const { return _can_seek(); }

        /// <summary>
        /// For any input stream, in_avail returns the number of bytes that are immediately available
        /// to be consumed without blocking. May be used in conjunction with sbumpc() to read data without
        /// incurring the overhead of using tasks.
        /// </summary>
        size_t in_avail() const { return _in_avail(); }

        bool is_open() const { return _is_open(); }

        /// <summary>
        /// Close the stream buffer, preventing further reade or write operations.
        /// </summary>
        pplx::task<bool> close() { return _close(); }

        /// <summary>
        /// Write a single byte to an output stream.
        /// </summary>
        /// <param name="ch">The byte to write</param>
        /// <returns>The value of the byte. -1 if the write operation fails</returns>
        pplx::task<int> aputc(byte ch) { return _aputc(ch); }

        /// <summary>
        /// Write a number of bytes to the stream.
        /// </summary>
        /// <param name="ptr">A pointer to the block of data to be written.</param>
        /// <param name="count">The number of bytes to write.</param>
        /// <returns>The number of bytes actually written, which may be less than what was requested. -1 if write the operation fails.</returns>
        pplx::task<size_t> aputn(const byte *ptr, size_t count) { return _aputn(ptr, count); }

        /// <summary>
        /// Copy a sequence of values from a collection given an iterator pair denoting the beginning and the end of the sequence.
        /// </summary>
        /// <param name="beg">The beginning of the sequence</param>
        /// <param name="end">The first position after the sequence</param>
        /// <returns>The number of bytes actually written. -1 if write the operation fails.</returns>
        /// <remarks>The max number of bytes is the sizeof(Elem) multiplied by the number of elements in the sequence.</remarks>
        template<typename InputIterator>
        pplx::task<size_t> copy_from(InputIterator beg, InputIterator end)
        {
            if ( beg == end || !can_write() ) return pplx::create_value<size_t>(0);

            auto total = new size_t;
            auto iter  = new InputIterator;
            auto val   = new InputIterator::value_type;

            *total = 0;
            *iter = beg;
            *val  = *beg;

            auto update = [=] (pplx::task<size_t> op) -> bool 
                { 
                    auto sz = op.get(); 
                    if ( sz == (size_t)-1 ) return false;
                    (*iter)++; 
                    (*total) += sz;
                    if ( *iter != end )
                        *val = **iter;
                    return *iter != end;
                };

            auto loop = pplx::do_while([=]() -> pplx::task<bool>
                {
                    return this->aputn((const byte *)val, sizeof(*val)).then(update);
                });

            return loop.then([=](pplx::task<bool> op) -> size_t
                { 
                    op.wait();
                    size_t sz = *total;
                    delete total; delete iter; delete val;
                    return sz; 
                });
        }

        /// <summary>
        /// Copy a sequence of values from a C-style array given pointers denoting the beginning and the first element after the sequence.
        /// </summary>
        /// <param name="beg">The beginning of the sequence</param>
        /// <param name="end">The first position after the sequence</param>
        /// <returns>The number of bytes actually written. -1 if write the operation fails.</returns>
        /// <remarks>The max number of bytes is the sizeof(Elem) multiplied by the number of elements in the sequence.</remarks>
        template<typename Elem>
        pplx::task<size_t> copy_from_ptr(const Elem *beg, const Elem * end)
        {
            if ( beg >= end || !can_write() ) return pplx::create_value<size_t>(0);

            auto total = new size_t;
            auto iter  = new (const Elem *);
            auto val   = new Elem;

            *total = 0;
            *iter = beg;
            *val  = *beg;

            auto update = [=] (pplx::task<size_t> op) -> bool 
                { 
                    auto sz = op.get(); 
                    if ( sz == (size_t)-1 ) return false;
                    (*iter)++; 
                    (*total) += sz;
                    if ( *iter < end )
                        *val = **iter;
                    return *iter < end;
                };

            auto loop = pplx::do_while([=]() -> pplx::task<bool>
                {
                    return this->aputn((const byte *)val, sizeof(*val)).then(update);
                });

            return loop.then([=](pplx::task<bool> op) -> size_t
                { 
                    op.wait();
                    size_t sz = *total;
                    delete total; delete iter; delete val;
                    return sz; 
                });
        }

        /// <summary>
        /// Write the given value to the stream the given number of times.
        /// </summary>
        /// <param name="num">The number of times the value should be written</param>
        /// <param name="value">The value to write</param>
        /// <returns>The total number of bytes written.</returns>
        /// <remarks>The max number of bytes is the sizeof(Elem) multiplied by the given number.</remarks>
        template<typename Elem>
        pplx::task<size_t> fill_n(size_t num, const Elem &value)
        {
            if ( num == 0 || !can_write() ) return pplx::create_value<size_t>(0);

            auto total = new size_t;
            auto count = new size_t;
            auto val   = new Elem;

            *total = 0;
            *count = num;
            *val = value;

            auto update = [=] (pplx::task<size_t> op) -> bool 
                { 
                    auto sz = op.get(); 
                    if ( sz == (size_t)-1 ) return false;
                    (*count)--; 
                    (*total) += sz; 
                    return *count > 0;
                };

            auto loop = pplx::do_while([=]() -> pplx::task<bool>
                {
                    return this->aputn((const byte *)val, sizeof(*val)).then(update);
                });

            return loop.then([=](pplx::task<bool> op) -> size_t
                { 
                    op.wait();
                    size_t sz = *total;
                    delete total; delete count; delete val;
                    return sz; 
                });
        }

        /// <summary>
        /// Call the given function the given number of times and write the return value to the stream.
        /// </summary>
        /// <param name="num">The number of times the function should be called</param>
        /// <param name="op">The function to call repeatedly</param>
        /// <returns>The total number of bytes written.</returns>
        /// <remarks>The max number of bytes is the sizeof(Elem) multiplied by the number of times to call.</remarks>
        template<typename Func>
        pplx::task<size_t> generate_n(size_t num, Func func)
        {
            if ( num == 0 || !can_write() ) return pplx::create_value<size_t>(0);

            auto total = new size_t;
            auto count = new size_t;
            auto val   = new decltype(func());

            *total = 0;
            *count = num;
            *val = func();

            auto update = [=] (pplx::task<size_t> op) -> bool 
                { 
                    auto sz = op.get(); 
                    if ( sz == (size_t)-1 ) return false;
                    (*count)--; 
                    (*total) += sz; 
                    if ( *count > 0 ) 
                        *val = func();
                    return *count > 0;
                };

            auto loop = pplx::do_while([=]() -> pplx::task<bool>
                {
                    return this->aputn((const byte *)val, sizeof(*val)).then(update);
                });

            return loop.then([=](pplx::task<bool> op) -> size_t
                { 
                    op.wait();
                    size_t sz = *total;
                    delete total; delete count; delete val;
                    return sz; 
                });
        }
        
        /// <summary>
        /// Read a single byte from the stream and advance the read position.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails.</returns>
        pplx::task<int> abumpc() { return _abumpc(); }

        /// <summary>
        /// Read a single byte from the stream and advance the read position.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails. -2 if an asynchronous read is required</returns>
        /// <remarks>This is a synchronous operation, but is guaranteed to never block.</remarks>
        int sbumpc() { return _sbumpc(); }

        /// <summary>
        /// Read a single byte from the stream without advancing the read position.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails.</returns>
        pplx::task<int> agetc() { return _agetc(); }

        /// <summary>
        /// Read a single byte from the stream without advancing the read position.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails. -2 if an asynchronous read is required</returns>
        /// <remarks>This is a synchronous operation, but is guaranteed to never block.</remarks>
        int sgetc() { return _sgetc(); }

        /// <summary>
        /// Advance the read position, then return the next character withouth advancing again.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails.</returns>
        pplx::task<int> anextc()  { return _anextc(); }

        /// <summary>
        /// Retreat the read position, then return the current character withouth advancing.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails. -2 if an asynchronous read is required</returns>
        pplx::task<int> aungetc() { return _aungetc(); }

        /// <summary>
        /// Read up to a given number of bytes from the stream.
        /// </summary>
        /// <param name="ptr">The address of the target memory area</param>
        /// <param name="count">The maximum number of bytes to read</param>
        /// <returns>The number of bytes read. O if the end of the stream is reached.</returns>
        pplx::task<size_t> agetn(byte *ptr, size_t count) { return _agetn(ptr, count); }

        /// <summary>
        /// Read up to a given number of bytes from the stream.
        /// </summary>
        /// <param name="ptr">The address of the target memory area</param>
        /// <param name="count">The maximum number of bytes to read</param>
        /// <returns>The number of bytes read. O if the end of the stream is reached or an asynchronous read is required.</returns>
        /// <remarks>This is a synchronous operation, but is guaranteed to never block.</remarks>
        size_t sgetn(byte *ptr, size_t count) { return _sgetn(ptr, count); }

        /// <summary>
        /// Copy data from the stream into a collection defined by an iterator pair.
        /// </summary>
        /// <param name="beg">The beginning of the target collection</param>
        /// <param name="end">The first position outside the target collection</param>
        /// <returns>The total number of bytes written. It may be less than what the iterator pair would indicate if the stream ends.</returns>
        template<typename ForwardIterator>
        pplx::task<size_t> copy_to(ForwardIterator beg, ForwardIterator end)           
        {
            if ( beg == end || !can_read() ) return pplx::create_value<size_t>(0);

            auto total = new size_t;
            auto iter  = new ForwardIterator;
            auto val   = new ForwardIterator::value_type;

            *total = 0;
            *iter = beg;
            memset(val, 0, sizeof(*beg));

            auto update = [=] (pplx::task<size_t> op) -> bool 
                { 
                    auto sz = op.get(); 
                    if ( sz == (size_t)-1 ) return false;
                    **iter = *val;
                    (*iter)++; 
                    (*total) += sz;
                    if ( *iter != end )
                        memset(val, 0, sizeof(*beg));
                    return *iter != end;
                };

            auto loop = pplx::do_while([=]() -> pplx::task<bool>
                {
                    return this->agetn((byte *)val, sizeof(*val)).then(update);
                });

            return loop.then([=](pplx::task<bool> op) -> size_t
                { 
                    op.wait();
                    size_t sz = *total;
                    delete total; delete iter; delete val;
                    return sz; 
                });
        }

        /// <summary>
        /// Copy data from the stream into a collection defined by a pair of pointers.
        /// </summary>
        /// <param name="beg">The beginning of the target collection</param>
        /// <param name="end">The first position outside the target collection</param>
        /// <returns>The total number of bytes written. It may be less than what the pointer pair would indicate if the stream ends.</returns>
        template<typename Elem>
        pplx::task<size_t> copy_to_ptr(Elem * beg, Elem * end)           
        {
            if ( beg >= end || !can_read() ) return pplx::create_value<size_t>(0);

            auto total = new size_t;
            auto iter  = new (Elem *);
            auto val   = new Elem;

            *total = 0;
            *iter = beg;
            memset(val, 0, sizeof(*beg));

            auto update = [=] (pplx::task<size_t> op) -> bool 
                { 
                    auto sz = op.get(); 
                    if ( sz == (size_t)-1 ) return false;
                    **iter = *val;
                    (*iter)++; 
                    (*total) += sz;
                    if ( *iter < end )
                        memset(val, 0, sizeof(*beg));
                    return *iter < end;
                };

            auto loop = pplx::do_while([=]() -> pplx::task<bool>
                {
                    return this->agetn((byte *)val, sizeof(*val)).then(update);
                });

            return loop.then([=](pplx::task<bool> op) -> size_t
                { 
                    op.wait();
                    size_t sz = *total;
                    delete total; delete iter; delete val;
                    return sz; 
                });
        }

        /// <summary>
        /// Seek to the given position.
        /// </summary>
        /// <param name="pos">The offset from the beginning of the stream</param>
        /// <param name="direction">The I/O direction to seek (see remarks)</param>
        /// <returns>The position. size_t(-1) if the operation fails.</returns>
        /// <remarks>Some streams may have separate write and read cursors. 
        ///          For such streams, the direction parameter defines whether to move the read or the write cursor.</remarks>
        pplx::task<size_t> aseekpos(size_t pos, std::ios_base::openmode direction) { return _aseekpos(pos, direction); }     

        /// <summary>
        /// Seek to a position given by a relative offset.
        /// </summary>
        /// <param name="offset">The relative position to seek to</param>
        /// <param name="way">The starting point (beginning, end, current) for the seek.</param>
        /// <param name="mode">The I/O direction to seek (see remarks)</param>
        /// <returns>The position. size_t(-1) if the operation fails.</returns>
        /// <remarks>Some streams may have separate write and read cursors. 
        ///          For such streams, the mode parameter defines whether to move the read or the write cursor.</remarks>
        pplx::task<size_t> aseekoff(ptrdiff_t offset, std::ios_base::seekdir way, std::ios_base::openmode mode) { return _aseekoff(offset, way, mode); }           

        /// <summary>
        /// For output streams, flush any internally buffered data to the underlying medium.
        /// </summary>
        /// <returns>O if the flush succeeds, -1 if not</returns>
        pplx::task<int> sync() { return _sync(); }

    protected:

        /// <summary>
        /// can_read is used to determine whether a stream buffer will support read operations (get).
        /// </summary>
        virtual bool _can_read() const = 0;

        /// <summary>
        /// can_write is used to determine whether a stream buffer will support write operations (put).
        /// </summary>       
        virtual bool _can_write() const = 0;

        /// <summary>
        /// can_seek is used to determine whether a stream buffer supports seeking.
        /// </summary>
        virtual bool _can_seek() const = 0;

        /// <summary>
        /// For any input stream, in_avail returns the number of bytes that are immediately available
        /// to be consumed without blocking. May be used in conjunction with sbumpc() to read data without
        /// incurring the overhead of using tasks.
        /// </summary>
        virtual size_t _in_avail() const = 0;

        virtual bool _is_open() const = 0;

        /// <summary>
        /// Close the stream buffer, preventing further reade or write operations.
        /// </summary>
        virtual pplx::task<bool> _close() = 0;

        /// <summary>
        /// Write a single byte to an output stream.
        /// </summary>
        /// <param name="ch">The byte to write</param>
        /// <returns>The value of the byte. -1 if the write operation fails</returns>
        virtual pplx::task<int> _aputc(byte ch) = 0;

        /// <summary>
        /// Write a number of bytes to the stream.
        /// </summary>
        /// <param name="ptr">A pointer to the block of data to be written.</param>
        /// <param name="count">The number of bytes to write.</param>
        /// <returns>The number of bytes actually written, which may be less than what was requested. -1 if write the operation fails.</returns>
        virtual pplx::task<size_t> _aputn(const byte *ptr, size_t count) = 0;

        /// <summary>
        /// Read a single byte from the stream and advance the read position.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails.</returns>
        virtual pplx::task<int> _abumpc() = 0;

        /// <summary>
        /// Read a single byte from the stream and advance the read position.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails. -2 if an asynchronous read is required</returns>
        /// <remarks>This is a synchronous operation, but is guaranteed to never block.</remarks>
        virtual int _sbumpc() = 0;

        /// <summary>
        /// Read a single byte from the stream without advancing the read position.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails.</returns>
        virtual pplx::task<int> _agetc() = 0;

        /// <summary>
        /// Read a single byte from the stream without advancing the read position.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails. -2 if an asynchronous read is required</returns>
        /// <remarks>This is a synchronous operation, but is guaranteed to never block.</remarks>
        virtual int _sgetc() = 0;

        /// <summary>
        /// Advance the read position, then return the next character withouth advancing again.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails.</returns>
        virtual pplx::task<int> _anextc() = 0;

        /// <summary>
        /// Retreat the read position, then return the current character withouth advancing.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails. -2 if an asynchronous read is required</returns>
        virtual pplx::task<int> _aungetc() = 0;

        /// <summary>
        /// Read up to a given number of bytes from the stream.
        /// </summary>
        /// <param name="ptr">The address of the target memory area</param>
        /// <param name="count">The maximum number of bytes to read</param>
        /// <returns>The number of bytes read. O if the end of the stream is reached.</returns>
        virtual pplx::task<size_t> _agetn(byte *ptr, size_t count) = 0;

        /// <summary>
        /// Read up to a given number of bytes from the stream.
        /// </summary>
        /// <param name="ptr">The address of the target memory area</param>
        /// <param name="count">The maximum number of bytes to read</param>
        /// <returns>The number of bytes read. O if the end of the stream is reached or an asynchronous read is required.</returns>
        /// <remarks>This is a synchronous operation, but is guaranteed to never block.</remarks>
        virtual size_t _sgetn(byte *ptr, size_t count) = 0;

        /// <summary>
        /// Seek to the given position.
        /// </summary>
        /// <param name="pos">The offset from the beginning of the stream</param>
        /// <param name="direction">The I/O direction to seek (see remarks)</param>
        /// <returns>The position. size_t(-1) if the operation fails.</returns>
        /// <remarks>Some streams may have separate write and read cursors. 
        ///          For such streams, the direction parameter defines whether to move the read or the write cursor.</remarks>
        virtual pplx::task<size_t> _aseekpos(size_t pos, std::ios_base::openmode direction) = 0;

        /// <summary>
        /// Seek to a position given by a relative offset.
        /// </summary>
        /// <param name="offset">The relative position to seek to</param>
        /// <param name="way">The starting point (beginning, end, current) for the seek.</param>
        /// <param name="mode">The I/O direction to seek (see remarks)</param>
        /// <returns>The position. size_t(-1) if the operation fails.</returns>
        /// <remarks>Some streams may have separate write and read cursors. 
        ///          For such streams, the mode parameter defines whether to move the read or the write cursor.</remarks>
        virtual pplx::task<size_t> _aseekoff(ptrdiff_t offset, std::ios_base::seekdir way, std::ios_base::openmode mode) = 0;           

        /// <summary>
        /// For output streams, flush any internally buffered data to the underlying medium.
        /// </summary>
        /// <returns>O if the flush succeeds, -1 if not</returns>
        virtual pplx::task<int> _sync() = 0;
    };

    /// <summary>
    /// Reference-counted stream buffer
    /// </summary>
    /// <remarks>
    /// The rationale for refcounting is discussed in the accompanying design
    /// documentation.
    /// </remarks>
    class streambuf : public basic_streambuf
    {
    public:
        /// <summary>
        /// Constructor
        /// </summary>
        /// <param name="ptr">A pointer to the concrete stream buffer implementation</param>
        streambuf(basic_streambuf *ptr) : m_buffer(ptr) { }

        /// <summary>
        /// Default constructor
        /// </summary>
        streambuf() { }

        /// <summary>
        /// Copy constructor
        /// </summary>
        /// <param name="other">The source object</param>
        streambuf(const streambuf &other) : m_buffer(other.m_buffer) { }

        /// <summary>
        /// Assignment operator
        /// </summary>
        /// <param name="other">The source object</param>
        streambuf & operator =(const streambuf &other) { m_buffer = other.m_buffer; return *this; }

        bool is_open() const { return m_buffer && m_buffer->is_open(); }

        /// <summary>
        /// Destructor
        /// </summary>
        virtual ~streambuf() { }

    protected:

#pragma region Function forwarding

        virtual bool _can_read() const { return m_buffer && m_buffer->can_read(); }
        virtual bool _can_write() const { return  m_buffer && m_buffer->can_write(); }
        virtual bool _can_seek() const { return m_buffer && m_buffer->can_seek(); }

        virtual size_t _in_avail() const { return m_buffer ? m_buffer->in_avail() : 0; }

        virtual bool _is_open() const { return m_buffer && m_buffer->is_open(); }

        virtual pplx::task<bool> _close()
        {
            return m_buffer ? m_buffer->close() : pplx::create_value(false); 
        }

        virtual pplx::task<int> _aputc(byte ch)
        {
            return m_buffer ? m_buffer->aputc(ch) : pplx::create_value<int>(-1); 
        }

        virtual pplx::task<size_t> _aputn(const byte *ptr, size_t count)
        {
            return m_buffer ? m_buffer->aputn(ptr, count) : pplx::create_value((size_t)-1); 
        }

        virtual pplx::task<int> _abumpc()
        {
            return m_buffer ? m_buffer->abumpc() : pplx::create_value<int>(-1); 
        }

        virtual int _sbumpc()
        {
            return m_buffer ? m_buffer->sbumpc() : -1; 
        }

        virtual pplx::task<int> _agetc()
        {
            return m_buffer ? m_buffer->agetc() : pplx::create_value<int>(-1); 
        }

        virtual int _sgetc()
        {
            return m_buffer ? m_buffer->sgetc() : -1; 
        }

        pplx::task<int> _anextc()
        {
            return m_buffer ? m_buffer->anextc() : pplx::create_value<int>(-1);  
        }

        pplx::task<int> _aungetc()
        {
            return m_buffer ? m_buffer->aungetc() : pplx::create_value<int>(-1);  
        }

        virtual pplx::task<size_t> _agetn(byte *ptr, size_t count)
        {
            return m_buffer ? m_buffer->agetn(ptr, count) : pplx::create_value<size_t>(0); 
        }

        virtual size_t _sgetn(byte *ptr, size_t count)
        {
            return m_buffer ? m_buffer->sgetn(ptr, count) : 0; 
        }

        virtual pplx::task<size_t> _aseekpos(size_t pos, std::ios_base::openmode direction)
        {
            return m_buffer ? m_buffer->aseekpos(pos, direction) : pplx::create_value((size_t)-1); 
        }

        virtual pplx::task<size_t> _aseekoff(ptrdiff_t offset, std::ios_base::seekdir way, std::ios_base::openmode mode)
        {
            return m_buffer ? m_buffer->aseekoff(offset, way, mode) : pplx::create_value((size_t)-1);
        }

        virtual pplx::task<int> _sync()
        {
            return m_buffer ? m_buffer->sync() : pplx::create_value<int>(-1);
        }
#pragma endregion

    private:
        std::shared_ptr<basic_streambuf> m_buffer;
    };

#if !defined(__cplusplus_winrt)
    /// <summary>
    /// Private stream buffer implementation for file streams.
    /// </summary>
    class basic_fstreambuf : public basic_streambuf
    {
    public:
        virtual ~basic_fstreambuf() { close(); }

    protected:

        /// <summary>
        /// can_read is used to determine whether a stream buffer will support read operations (get).
        /// </summary>
        virtual bool _can_read() const { return _is_open() && (m_info->m_mode & std::ios_base::in) != 0; }
        
        /// <summary>
        /// can_write is used to determine whether a stream buffer will support write operations (put).
        /// </summary>       
        virtual bool _can_write() const { return _is_open() && (m_info->m_mode & std::ios_base::out) != 0; }

        /// <summary>
        /// can_seek is used to determine whether a stream buffer supports seeking.
        /// </summary>
        virtual bool _can_seek() const { return _is_open(); }

        /// <summary>
        /// For any input stream, in_avail returns the number of bytes that are immediately available
        /// to be consumed without blocking. May be used in conjunction with sbumpc() to read data without
        /// incurring the overhead of using tasks.
        /// </summary>
        virtual size_t _in_avail() const
        {
            actors::details::win32::reentrant_lock::scoped_lock lck(m_info->m_lock);

            return _in_avail_unprot();
        }

        size_t _in_avail_unprot() const
        {
            if ( !_is_open() ) return 0;

            if ( m_info->m_buffer == nullptr || m_info->m_buffill == 0 ) return 0;

            size_t bufpos = m_info->m_rdpos - m_info->m_bufoff;
            return m_info->m_buffill - bufpos;
        }

        virtual bool _is_open() const
        {
            return m_info != nullptr; 
        }

        /// <summary>
        /// Close the stream buffer, preventing further reade or write operations.
        /// </summary>
        virtual pplx::task<bool> _close()
        {
            if ( !_is_open() ) return pplx::create_value(false);

            if ( _can_write() )
            {
                // We need to flush all writes if the file was opened for writing.

                auto s = _sync();

                return s.then(
                    [=](pplx::task<int> op) -> pplx::task<bool>
                    {
                        int sync_res = op.get();

                        if ( sync_res != 0 )
                            return pplx::create_value(false);

                        auto result_tce = pplx::task_completion_event<bool>();
                        auto callback = new _filestream_callback_close(result_tce);

                        // The context when called from the continuation of sync() is that
                        // 
                        if ( !_close_fsb_nolock(&m_info, callback) )
                        {
                            delete callback;
                            return pplx::create_value(false);
                        }
                        return pplx::task<bool>(result_tce);
                    });
            }
            else
            {
                auto result_tce = pplx::task_completion_event<bool>();
                auto callback = new _filestream_callback_close(result_tce);
                if ( !_close_fsb(&m_info, callback) )
                {
                    delete callback;
                    return pplx::create_value(false);
                }
                return pplx::task<bool>(result_tce);
            }
        }

        /// <summary>
        /// Write a single byte to an output stream.
        /// </summary>
        /// <param name="ch">The byte to write</param>
        /// <returns>The value of the byte. -1 if the write operation fails</returns>
        virtual pplx::task<int> _aputc(byte ch)
        {
            if ( !_can_write() )
                return pplx::create_value<int>(-1);

            auto result_tce = pplx::task_completion_event<int>();
            auto callback = new _filestream_callback_aputc(m_info, result_tce, ch);
             
            size_t written = _aputc_fsb(m_info, callback, ch);

            switch (written)
            {
            case -1:
                delete callback;
                return pplx::create_value<int>(-1);
            case sizeof(byte):
                {
                    delete callback;
                    actors::details::win32::reentrant_lock::scoped_lock lck(m_info->m_lock);
                    if ( ((int)m_info->m_wrpos) >= 0 ) m_info->m_wrpos += written;
                    return pplx::create_value<int>(ch);
                }
                break;
            }

            return pplx::task<int>(result_tce);
        }

        /// <summary>
        /// Write a number of bytes to the stream.
        /// </summary>
        /// <param name="ptr">A pointer to the block of data to be written.</param>
        /// <param name="count">The number of bytes to write.</param>
        /// <returns>The number of bytes actually written, which may be less than what was requested. -1 if write the operation fails.</returns>
        virtual pplx::task<size_t> _aputn(const byte *ptr, size_t count)
        {
            if ( !_can_write() )
                return pplx::create_value((size_t)-1);

            auto result_tce = pplx::task_completion_event<size_t>();
            auto callback = new _filestream_callback_write<size_t>(m_info, result_tce);

            size_t written = _aputn_fsb(m_info, callback, ptr, count);

            if ( written != 0 )
            {
                delete callback;
                if ( written != -1 )
                {
                    actors::details::win32::reentrant_lock::scoped_lock lck(m_info->m_lock);
                    if ( ((int)m_info->m_wrpos) >= 0 ) m_info->m_wrpos += written;
                }
                return pplx::create_value<size_t>(written);
            }
            return pplx::task<size_t>(result_tce);
        }

        /// <summary>
        /// Read a single byte from the stream and advance the read position.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails.</returns>
        virtual pplx::task<int> _abumpc()
        {
            if ( !_can_read() )
                return pplx::create_value<int>(-1);

            if ( _in_avail_unprot() > 0 )
            {
                actors::details::win32::reentrant_lock::scoped_lock lck(m_info->m_lock);

                // Check again once the lock is held.

                if ( _in_avail_unprot() > 0 )
                {
                    auto bufoff = m_info->m_rdpos - m_info->m_bufoff;
                    byte ch = m_info->m_buffer[bufoff];
                    m_info->m_rdpos += 1;
                    return pplx::create_value<int>(ch);
                }
            }

            auto result_tce = pplx::task_completion_event<int>();
            auto callback = new _filestream_callback_abumpc(m_info, result_tce);

            size_t ch = _agetn_fsb(m_info, callback, &callback->m_ch, 1);

            switch (ch)
            {
            case (size_t)-1:
                {
                    delete callback;
                    return pplx::create_value<int>(-1);
                }
                break;
            case sizeof(byte):
                {
                    actors::details::win32::reentrant_lock::scoped_lock lck(m_info->m_lock);
                    m_info->m_rdpos += 1;
                    auto ch1 = callback->m_ch;
                    delete callback;
                    return pplx::create_value<int>(ch1);
                }
                break;
            }

            return pplx::task<int>(result_tce);
        }

        /// <summary>
        /// Read a single byte from the stream and advance the read position.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails. -2 if an asynchronous read is required</returns>
        /// <remarks>This is a synchronous operation, but is guaranteed to never block.</remarks>
        virtual int _sbumpc() 
        {
            if ( m_info->m_atend || !_can_read() ) return -1;

            if ( _in_avail_unprot() == 0 ) return -2;

            actors::details::win32::reentrant_lock::scoped_lock lck(m_info->m_lock);

            if ( _in_avail_unprot() == 0 ) return -2;

            auto bufoff = m_info->m_rdpos - m_info->m_bufoff;
            char ch = m_info->m_buffer[bufoff];
            m_info->m_rdpos += 1;
            return ch;
        }

        /// <summary>
        /// Read a single byte from the stream without advancing the read position.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails.</returns>
        pplx::task<int> _agetc()
        {
            if ( !_can_read() )
                return pplx::create_value<int>(-1);

            if ( _in_avail_unprot() > 0 )
            {
                actors::details::win32::reentrant_lock::scoped_lock lck(m_info->m_lock);

                // Check again once the lock is held.

                if ( _in_avail_unprot() > 0 )
                {
                    auto bufoff = m_info->m_rdpos - m_info->m_bufoff;
                    byte ch = m_info->m_buffer[bufoff];
                    return pplx::create_value<int>(ch);
                }
            }

            auto result_tce = pplx::task_completion_event<int>();
            auto callback = new _filestream_callback_agetc(m_info, result_tce);

            size_t ch = _agetn_fsb(m_info, callback, &callback->m_ch, 1);

            switch (ch)
            {
            case (size_t)-1:
                {
                    delete callback;
                    return pplx::create_value<int>(-1);
                }
                break;
            case sizeof(byte):
                {
                    actors::details::win32::reentrant_lock::scoped_lock lck(m_info->m_lock);
                    auto ch1 = callback->m_ch;
                    delete callback;
                    return pplx::create_value<int>(ch1);
                }
                break;
            }

            return pplx::task<int>(result_tce);
        }

        /// <summary>
        /// Read a single byte from the stream without advancing the read position.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails. -2 if an asynchronous read is required</returns>
        /// <remarks>This is a synchronous operation, but is guaranteed to never block.</remarks>
        int _sgetc()
        {
            if ( m_info->m_atend || !_can_read() ) return -1;

            if ( _in_avail_unprot() == 0 ) return -2;

            actors::details::win32::reentrant_lock::scoped_lock lck(m_info->m_lock);

            if ( _in_avail_unprot() == 0 ) return -2;

            auto bufoff = m_info->m_rdpos - m_info->m_bufoff;
            char ch = m_info->m_buffer[bufoff];
            return ch;
        }

        /// <summary>
        /// Advance the read position, then return the next character withouth advancing again.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails.</returns>
        virtual pplx::task<int> _anextc()
        {
            _seekrdpos_fsb(m_info, m_info->m_rdpos+1);
            if ( m_info->m_atend )
                return pplx::create_value<int>(-1);
            return agetc();
        }

        /// <summary>
        /// Retreat the read position, then return the current character withouth advancing.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails. -2 if an asynchronous read is required</returns>
        virtual pplx::task<int> _aungetc() 
        {
            if ( m_info->m_rdpos == 0 ) 
                return pplx::create_value<int>(-1);
            _seekrdpos_fsb(m_info, m_info->m_rdpos-1);
            return agetc();
        }

        /// <summary>
        /// Read up to a given number of bytes from the stream.
        /// </summary>
        /// <param name="ptr">The address of the target memory area</param>
        /// <param name="count">The maximum number of bytes to read</param>
        /// <returns>The number of bytes read. O if the end of the stream is reached, -1 if there's some error.</returns>
        virtual pplx::task<size_t> _agetn(byte *ptr, size_t count)
        {
            if ( !_can_read() )
                return pplx::create_value((size_t)-1);
            if ( m_info->m_atend )
                return pplx::create_value<size_t>(0);

            if ( _in_avail_unprot() >= count )
            {
                actors::details::win32::reentrant_lock::scoped_lock lck(m_info->m_lock);

                // Check again once the lock is held.

                if ( _in_avail_unprot() >= count )
                {
                    auto bufoff = m_info->m_rdpos - m_info->m_bufoff;
                    memcpy((void *)ptr, this->m_info->m_buffer+bufoff, count);
            
                    m_info->m_rdpos += count;
                    return pplx::create_value<size_t>(count);
                }
            }

            auto result_tce = pplx::task_completion_event<size_t>();
            auto callback = new _filestream_callback_read(m_info, result_tce);

            size_t read = _agetn_fsb(m_info, callback, ptr, count);

            if ( read != 0 )
            {
                delete callback;
                if ( read != -1 )
                {
                    actors::details::win32::reentrant_lock::scoped_lock lck(m_info->m_lock);
                    m_info->m_rdpos += read;
                }
                return pplx::create_value<size_t>(read);
            }
            return pplx::task<size_t>(result_tce);
        }

        /// <summary>
        /// Read up to a given number of bytes from the stream.
        /// </summary>
        /// <param name="ptr">The address of the target memory area</param>
        /// <param name="count">The maximum number of bytes to read</param>
        /// <returns>The number of bytes read. O if the end of the stream is reached or an asynchronous read is required.</returns>
        /// <remarks>This is a synchronous operation, but is guaranteed to never block.</remarks>
        virtual size_t _sgetn(byte *ptr, size_t count)
        {
            if ( m_info->m_atend || !_can_read() ) return 0;

            if ( _in_avail() == 0 ) return 0;

            actors::details::win32::reentrant_lock::scoped_lock lck(m_info->m_lock);

            size_t available = _in_avail_unprot();
            size_t copy = (count < available) ? count : available;

            auto bufoff = m_info->m_rdpos - m_info->m_bufoff;
            memcpy((void *)ptr, this->m_info->m_buffer+bufoff, copy);
            
            m_info->m_rdpos += copy;
            m_info->m_atend = (copy < count);
            return copy;
        }

        /// <summary>
        /// Seek to the given position.
        /// </summary>
        /// <param name="pos">The offset from the beginning of the stream</param>
        /// <param name="direction">The I/O direction to seek (see remarks)</param>
        /// <returns>The position. size_t(-1) if the operation fails.</returns>
        /// <remarks>Some streams may have separate write and read cursors. 
        ///          For such streams, the direction parameter defines whether to move the read or the write cursor.</remarks>
        virtual pplx::task<size_t> _aseekpos(size_t pos, std::ios_base::openmode mode) 
        {
            if ( mode == std::ios_base::in ) 
            {
                return pplx::create_value(_seekrdpos_fsb(m_info, pos)); 
            }
            else if ( (m_info->m_mode & std::ios::ios_base::app) == 0 )
            {
                return pplx::create_value(_seekwrpos_fsb(m_info, pos)); 
            }
            return pplx::create_value((size_t)-1); 
        }

        /// <summary>
        /// Seek to a position given by a relative offset.
        /// </summary>
        /// <param name="offset">The relative position to seek to</param>
        /// <param name="way">The starting point (beginning, end, current) for the seek.</param>
        /// <param name="mode">The I/O direction to seek (see remarks)</param>
        /// <returns>The position. size_t(-1) if the operation fails.</returns>
        /// <remarks>Some streams may have separate write and read cursors. 
        ///          For such streams, the mode parameter defines whether to move the read or the write cursor.</remarks>
        virtual pplx::task<size_t> _aseekoff(ptrdiff_t offset, std::ios_base::seekdir way, std::ios_base::openmode mode) 
        {
            if ( mode == std::ios_base::in ) 
            {
                switch ( way )
                {
                case std::ios_base::beg:
                    return pplx::create_value(_seekrdpos_fsb(m_info, offset));
                case std::ios_base::cur:
                    return pplx::create_value(_seekrdpos_fsb(m_info, m_info->m_rdpos+offset));
                case std::ios_base::end:
                    break;
                }
            }
            else if ( (m_info->m_mode & std::ios::ios_base::app) == 0 )
            {
                switch ( way )
                {
                case std::ios_base::beg:
                    return pplx::create_value(_seekwrpos_fsb(m_info, offset));
                case std::ios_base::cur:
                    return pplx::create_value(_seekwrpos_fsb(m_info, m_info->m_wrpos+offset));
                case std::ios_base::end:
                    return pplx::create_value(_seekwrpos_fsb(m_info, (size_t)-1));
                    break;
                }
            }
            return pplx::create_value((size_t)-1); 
        }

        /// <summary>
        /// For output streams, flush any internally buffered data to the underlying medium.
        /// </summary>
        /// <returns>O if the flush succeeds, -1 if not</returns>
        virtual pplx::task<int> _sync()
        {
            if ( !_can_write() )
                return pplx::create_value<int>(-1);

            auto result_tce = pplx::task_completion_event<int>();
            auto callback = new _filestream_callback_write<int>(m_info, result_tce);

            if ( !_sync_fsb(m_info, callback) )
            {
                delete callback;
                return pplx::create_value<int>(-1);
            }
            return pplx::task<int>(result_tce);
        }

    private:
        friend class fstreambuf;
        friend basic_fstreambuf *async::details::_createfsb(details::_file_info *);

        basic_fstreambuf(details::_file_info *info) : m_info(info) { }

        static pplx::task<basic_streambuf*> open(
            const std::string &_Filename,
            std::ios_base::openmode _Mode = std::ios_base::out,
            int _Prot = (int)std::ios_base::_Openprot)
        {
            auto result_tce = pplx::task_completion_event<basic_streambuf*>();
            auto callback = new _filestream_callback_open(result_tce);
            if ( !_open_fsb_str(callback, _Filename.c_str(), _Mode, _Prot) )
            {
                delete callback;
                return pplx::create_value<basic_streambuf*>(nullptr);
            }
            return pplx::task<basic_streambuf*>(result_tce);
        }

        static pplx::task<basic_streambuf*> open(
            const std::wstring &_Filename,
            std::ios_base::openmode _Mode = std::ios_base::out,
            int _Prot = (int)std::ios_base::_Openprot)
        {
            auto result_tce = pplx::task_completion_event<basic_streambuf*>();
            auto callback = new _filestream_callback_open(result_tce);
            if ( !_open_fsb_wstr(callback, _Filename.c_str(), _Mode, _Prot) )
            {
                delete callback;
                return pplx::create_value<basic_streambuf*>(nullptr);
            }
            return pplx::task<basic_streambuf*>(result_tce);
        }

#pragma region Completion callback interface implementations
        class _filestream_callback_open : public details::_filestream_callback
        {
        public:
            _filestream_callback_open(pplx::task_completion_event<basic_streambuf*> op) : m_op(op) { }

            virtual void on_opened(basic_streambuf *buf)
            {
                m_op.set(buf);
                delete this;
            }

        private:
            pplx::task_completion_event<basic_streambuf*> m_op;
        };

        class _filestream_callback_close : public details::_filestream_callback
        {
        public:
            _filestream_callback_close(pplx::task_completion_event<bool> op) : m_op(op) { }

            virtual void on_closed(bool result)
            {
                m_op.set(result);
                delete this;
            }

        private:
            pplx::task_completion_event<bool> m_op;
        };

        template<typename ResultType>
        class _filestream_callback_write : public details::_filestream_callback
        {
        public:
            _filestream_callback_write(details::_file_info *info, pplx::task_completion_event<ResultType> op) : m_info(info), m_op(op) { }

            virtual void on_completed(size_t result)
            {
                if ( ((int)m_info->m_wrpos) >= 0 ) m_info->m_wrpos += result;
                m_op.set((ResultType)result);
                delete this;
            }

        private:
            details::_file_info *m_info;
            pplx::task_completion_event<ResultType> m_op;
        };

        class _filestream_callback_read : public details::_filestream_callback
        {
        public:
            _filestream_callback_read(details::_file_info *info, pplx::task_completion_event<size_t> op) : m_op(op), m_info(info) { }

            virtual void on_completed(size_t result)
            {
                m_info->m_rdpos += result;
                m_op.set(result);
                delete this;
            }

        private:
            details::_file_info *m_info;
            pplx::task_completion_event<size_t> m_op;
        };

        class _filestream_callback_aputc : public details::_filestream_callback
        {
        public:
            _filestream_callback_aputc(details::_file_info *info, pplx::task_completion_event<int> op, byte ch) : m_info(info), m_ch(ch), m_op(op) { }

            virtual void on_completed(size_t result)
            {
                if ( result == sizeof(byte) )
                {
                    if ( ((int)m_info->m_wrpos) >= 0 ) m_info->m_wrpos += 1;
                    m_op.set(m_ch);   
                }
                else
                {
                    m_op.set(-1);
                }
                delete this;
            }

        private:
            details::_file_info *m_info;
            pplx::task_completion_event<int> m_op;
            byte           m_ch;
        };

        class _filestream_callback_abumpc : public details::_filestream_callback
        {
        public:
            _filestream_callback_abumpc(details::_file_info *info, pplx::task_completion_event<int> op) : m_op(op), m_info(info) { }

            virtual void on_completed(size_t result)
            {
                if ( result == sizeof(byte) )
                {
                    m_info->m_rdpos += 1;
                    m_op.set(m_ch);   
                }
                else
                {
                    m_op.set(-1);
                }
                delete this;
            }

            byte           m_ch;

        private:
            details::_file_info *m_info;
            pplx::task_completion_event<int> m_op;
        };

        class _filestream_callback_agetc : public details::_filestream_callback
        {
        public:
            _filestream_callback_agetc(details::_file_info *info, pplx::task_completion_event<int> op) : m_op(op), m_info(info) { }

            virtual void on_completed(size_t result)
            {
                if ( result == sizeof(byte) )
                {
                    m_op.set(m_ch);   
                }
                else
                {
                    m_op.set(-1);
                }
                delete this;
            }

            byte           m_ch;

        private:
            details::_file_info *m_info;
            pplx::task_completion_event<int> m_op;
        };
#pragma endregion

        details::_file_info *m_info;
    };

    /// <summary>
    /// Stream buffer for file streams.
    /// </summary>
    class fstreambuf : public streambuf
    {
    public:
        /// <summary>
        /// Open a new stream buffer representing the given file.
        /// </summary>
        /// <param name="file_name">The name of the file</param>
        /// <param name="mode">The opening mode of the file</param>
        /// <param name="prot">The file protection mode</param>
        /// <returns>An opened stream buffer.</returns>
        static pplx::task<streambuf> open(
            const std::string &file_name,
            std::ios_base::openmode mode = std::ios_base::out,
            int prot = (int)std::ios_base::_Openprot)
        {
            auto bfb = basic_fstreambuf::open(file_name, mode, prot);
            return bfb.then([](pplx::task<basic_streambuf *> op) -> streambuf
                {
                    return streambuf(op.get());
                });
        }

        /// <summary>
        /// Open a new stream buffer representing the given file.
        /// </summary>
        /// <param name="file_name">The name of the file</param>
        /// <param name="mode">The opening mode of the file</param>
        /// <param name="prot">The file protection mode</param>
        /// <returns>An opened stream buffer.</returns>
        static pplx::task<streambuf> open(
            const std::wstring &file_name,
            std::ios_base::openmode mode = std::ios_base::out,
            int prot = (int)std::ios_base::_Openprot)
        {
            auto bfb = basic_fstreambuf::open(file_name, mode, prot);
            return bfb.then([](pplx::task<basic_streambuf *> op) -> streambuf
                {
                    return streambuf(op.get());
                });
        }
    };

    /// <summary>
    /// Private stream buffer implementation for tcp sockets.
    /// </summary>
    class basic_tcpstreambuf : public basic_streambuf
    {
    public:
        virtual ~basic_tcpstreambuf()
        { 
            close().get(); 
        }

    protected:

        /// <summary>
        /// can_read is used to determine whether a stream buffer will support read operations (get).
        /// </summary>
        virtual bool _can_read() const { return true; }

        /// <summary>
        /// can_write is used to determine whether a stream buffer will support write operations (put).
        /// </summary>       
        virtual bool _can_write() const { return true; }
        
        /// <summary>
        /// can_seek is used to determine whether a stream buffer supports seeking.
        /// </summary>
        virtual bool _can_seek() const { return false; }

        /// <summary>
        /// For any input stream, in_avail returns the number of bytes that are immediately available
        /// to be consumed without blocking. May be used in conjunction with sbumpc() to read data without
        /// incurring the overhead of using tasks.
        /// </summary>
        virtual size_t _in_avail() const { return 0; }

        virtual bool _is_open() const
        {
            return m_info != nullptr; 
        }

        /// <summary>
        /// Close the stream buffer, preventing further reade or write operations.
        /// </summary>
        virtual pplx::task<bool> _close()
        {
            if ( !_is_open() ) return pplx::create_value(false);

            if ( _can_write() )
            {
                // We need to flush all writes if the file was opened for writing.
                auto s = _sync();
                return s.then(
                    [=](pplx::task<int> op) -> pplx::task<bool>
                    {
                        int sync_res = op.get();

                        if ( sync_res != 0 )
                            return pplx::create_value(false);

                        auto result_tce = pplx::task_completion_event<bool>();
                        auto callback = new _tcpstream_callback_close(result_tce);
                        if ( !_close_tcp(&m_info, callback) )
                        {
                            delete callback;
                            return pplx::create_value(false);
                        }
                        return pplx::task<bool>(result_tce);
                    });
            }
            else
            {
                auto result_tce = pplx::task_completion_event<bool>();
                auto callback = new _tcpstream_callback_close(result_tce);
                if ( !_close_tcp(&m_info, callback) )
                {
                    delete callback;
                    return pplx::create_value(false);
                }
                return pplx::task<bool>(result_tce);
            }
        } 

        /// <summary>
        /// Write a single byte to an output stream.
        /// </summary>
        /// <param name="ch">The byte to write</param>
        /// <returns>The value of the byte. -1 if the write operation fails</returns>
        virtual pplx::task<int> _aputc(byte ch)
        {
            if ( !_can_write() )
                return pplx::create_value(-1);

            auto result_tce = pplx::task_completion_event<int>();
            auto callback = new _tcpstream_callback_aputc(m_info, result_tce, ch);

            size_t written = _aputc_tcp(m_info, callback, ch);

            switch (written)
            {
            case -1:
                delete callback;
                return pplx::create_value(-1);            
            case sizeof(byte):
                {
                    delete callback;
                    return pplx::create_value<int>(ch);
                }
                break;
            }

            return pplx::task<int>(result_tce);
        }

        /// <summary>
        /// Write a number of bytes to the stream.
        /// </summary>
        /// <param name="ptr">A pointer to the block of data to be written.</param>
        /// <param name="count">The number of bytes to write.</param>
        /// <returns>The number of bytes actually written, which may be less than what was requested. -1 if write the operation fails.</returns>
        virtual pplx::task<size_t> _aputn(const byte *ptr, size_t count)
        {
            if ( !_can_write() )
                return pplx::create_value((size_t)-1);

            auto result_tce = pplx::task_completion_event<size_t>();
            auto callback = new _tcpstream_callback_write<size_t>(m_info, result_tce);

            size_t written = _aputn_tcp(m_info, callback, ptr, count);

            if ( written != 0 )
            {
                delete callback;
                if ( written != -1 )
                {
                    actors::details::win32::reentrant_lock::scoped_lock lck(m_info->m_lock);
                }
                return pplx::create_value(written);
            }
            return pplx::task<size_t>(result_tce);
        }

        /// <summary>
        /// Read a single byte from the stream and advance the read position.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails.</returns>
        virtual pplx::task<int> _abumpc()
        {
            if ( !_can_read() )
                return pplx::create_value(-1);

            auto result_tce = pplx::task_completion_event<int>();
            auto callback = new _tcpstream_callback_abumpc(m_info, result_tce);

            size_t ch = _agetn_tcp(m_info, callback, &callback->m_ch, 1);

            switch (ch)
            {
            case (size_t)-1:
                {
                    delete callback;
                    return pplx::create_value(-1);
                }
                break;
            case sizeof(byte):
                {
                    actors::details::win32::reentrant_lock::scoped_lock lck(m_info->m_lock);
                    auto ch1 = callback->m_ch;
                    delete callback;
                    return pplx::create_value<int>(ch1);
                }
                break;
            }

            return pplx::task<int>(result_tce);
        }

        /// <summary>
        /// Read a single byte from the stream and advance the read position.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails. -2 if an asynchronous read is required</returns>
        /// <remarks>This is a synchronous operation, but is guaranteed to never block.</remarks>
        virtual int _sbumpc() { return -2; }

        /// <summary>
        /// Read a single byte from the stream without advancing the read position.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails.</returns>
        virtual pplx::task<int> _agetc() { return pplx::create_value(-1); }

        /// <summary>
        /// Read a single byte from the stream without advancing the read position.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails. -2 if an asynchronous read is required</returns>
        /// <remarks>This is a synchronous operation, but is guaranteed to never block.</remarks>
        virtual int _sgetc() { return -2; }

        /// <summary>
        /// Advance the read position, then return the next character withouth advancing again.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails.</returns>
        virtual pplx::task<int> _anextc() { return pplx::create_value(-1); }

        /// <summary>
        /// Retreat the read position, then return the current character withouth advancing.
        /// </summary>
        /// <returns>The value of the byte. -1 if the read fails. -2 if an asynchronous read is required</returns>
        virtual pplx::task<int> _aungetc() { return pplx::create_value(-1); }

        /// <summary>
        /// Read up to a given number of bytes from the stream.
        /// </summary>
        /// <param name="ptr">The address of the target memory area</param>
        /// <param name="count">The maximum number of bytes to read</param>
        /// <returns>The number of bytes read. O if the end of the stream is reached, -1 if there's some error.</returns>
        virtual pplx::task<size_t> _agetn(byte *ptr, size_t count)
        {
            auto result_tce = pplx::task_completion_event<size_t>();
            auto callback = new _tcpstream_callback_read(m_info, result_tce);

            size_t read = _agetn_tcp(m_info, callback, ptr, count);

            if ( read != 0 )
            {
                delete callback;
                return pplx::create_value(read);
            }
            return pplx::task<size_t>(result_tce);
        }

        /// <summary>
        /// Read up to a given number of bytes from the stream.
        /// </summary>
        /// <param name="ptr">The address of the target memory area</param>
        /// <param name="count">The maximum number of bytes to read</param>
        /// <returns>The number of bytes read. O if the end of the stream is reached or an asynchronous read is required.</returns>
        /// <remarks>This is a synchronous operation, but is guaranteed to never block.</remarks>
        virtual size_t _sgetn(byte *ptr, size_t count) { return 0; }

        /// <summary>
        /// Seek to the given position.
        /// </summary>
        /// <param name="pos">The offset from the beginning of the stream</param>
        /// <param name="direction">The I/O direction to seek (see remarks)</param>
        /// <returns>The position. size_t(-1) if the operation fails.</returns>
        /// <remarks>Some streams may have separate write and read cursors. 
        ///          For such streams, the direction parameter defines whether to move the read or the write cursor.</remarks>
        virtual pplx::task<size_t> _aseekpos(size_t pos, std::ios_base::openmode direction) { return pplx::create_value((size_t)-1); }

        /// <summary>
        /// Seek to a position given by a relative offset.
        /// </summary>
        /// <param name="offset">The relative position to seek to</param>
        /// <param name="way">The starting point (beginning, end, current) for the seek.</param>
        /// <param name="mode">The I/O direction to seek (see remarks)</param>
        /// <returns>The position. size_t(-1) if the operation fails.</returns>
        /// <remarks>Some streams may have separate write and read cursors. 
        ///          For such streams, the mode parameter defines whether to move the read or the write cursor.</remarks>
        virtual pplx::task<size_t> _aseekoff(ptrdiff_t offset, std::ios_base::seekdir way, std::ios_base::openmode mode) { return pplx::create_value((size_t)-1); }           

        /// <summary>
        /// For output streams, flush any internally buffered data to the underlying medium.
        /// </summary>
        /// <returns>O if the flush succeeds, -1 if not</returns>
        /// <remarks>Sync on a TCP stream buffer simply means to wait for all outstanding writes to finish, there
        ///          is no interal stream buffering.
        /// </remarks>
        virtual pplx::task<int> _sync()
        {
            if ( !_can_write() )
                return pplx::create_value(-1);

            auto result_tce = pplx::task_completion_event<int>();
            auto callback = new _tcpstream_callback_write<int>(m_info, result_tce);

            if ( !_sync_tcp(m_info, callback) )
            {
                delete callback;
                return pplx::create_value(-1);
            }
            return pplx::task<int>(result_tce);
        }

    private:
        friend class tcpstreambuf;
        friend class tcp_listener;
        friend basic_tcpstreambuf * details::_createtcp(details::_tcp_info *);

        basic_tcpstreambuf(details::_tcp_info *info) : m_info(info) { }

        static pplx::task<basic_streambuf*> connect(
            const std::string &host, 
            const std::string &service)
        {
            auto result_tce = pplx::task_completion_event<basic_streambuf*>();
            auto callback = new _tcpstream_callback_open(result_tce);
            if ( !_connect_tcp_str(callback, host.c_str(), service.c_str()) )
            {
                delete callback;
                return pplx::create_value<basic_streambuf*>(nullptr);
            }
            return pplx::task<basic_streambuf*>(result_tce);
        }

        static pplx::task<basic_streambuf*> connect(
            const std::wstring &host, 
            const std::wstring &service)
        {
            auto result_tce = pplx::task_completion_event<basic_streambuf*>();
            auto callback = new _tcpstream_callback_open(result_tce);
            if ( !_connect_tcp_wstr(callback, host.c_str(), service.c_str()) )
            {
                delete callback;
                return pplx::create_value<basic_streambuf*>(nullptr);
            }
            return pplx::task<basic_streambuf*>(result_tce);
        }

#pragma region Completion callback interface implementations
        class _tcpstream_callback_open : public details::_filestream_callback
        {
        public:
            _tcpstream_callback_open(pplx::task_completion_event<basic_streambuf*> op) : m_op(op) { }

            virtual void on_opened(basic_streambuf *buf)
            {
                m_op.set(buf);
                delete this;
            }

        private:
            pplx::task_completion_event<basic_streambuf*> m_op;
        };

        class _tcpstream_callback_close : public details::_filestream_callback
        {
        public:
            _tcpstream_callback_close(pplx::task_completion_event<bool> op) : m_op(op) { }

            virtual void on_closed(bool result)
            {
                m_op.set(result);
                delete this;
            }

        private:
            pplx::task_completion_event<bool> m_op;
        };

        class _tcpstream_callback_read : public details::_filestream_callback
        {
        public:
            _tcpstream_callback_read(details::_tcp_info *info, pplx::task_completion_event<size_t> op) : m_op(op), m_info(info) { }

            virtual void on_completed(size_t result)
            {
                m_op.set(result);
                delete this;
            }

        private:
            details::_tcp_info *m_info;
            pplx::task_completion_event<size_t> m_op;
        };

        template<typename ResultType>
        class _tcpstream_callback_write : public details::_filestream_callback
        {
        public:
            _tcpstream_callback_write(details::_tcp_info *info, pplx::task_completion_event<ResultType> op) : m_info(info), m_op(op) { }

            virtual void on_completed(size_t result)
            {
                m_op.set((ResultType)result);
                delete this;
            }

        private:
            details::_tcp_info *m_info;
            pplx::task_completion_event<ResultType> m_op;
        };

        class _tcpstream_callback_aputc : public details::_filestream_callback
        {
        public:
            _tcpstream_callback_aputc(details::_tcp_info *info, pplx::task_completion_event<int> op, byte ch) : m_info(info), m_ch(ch), m_op(op) { }

            virtual void on_completed(size_t result)
            {
                if ( result == sizeof(byte) )
                {
                    m_op.set(m_ch);   
                }
                else
                {
                    m_op.set(-1);
                }
                delete this;
            }

        private:
            details::_tcp_info *m_info;
            pplx::task_completion_event<int> m_op;
            byte           m_ch;
        };

        class _tcpstream_callback_abumpc : public details::_filestream_callback
        {
        public:
            _tcpstream_callback_abumpc(details::_tcp_info *info, pplx::task_completion_event<int> op) : m_op(op), m_info(info) { }

            virtual void on_completed(size_t result)
            {
                if ( result == sizeof(byte) )
                {
                    m_op.set(m_ch);   
                }
                else
                {
                    m_op.set(-1);
                }
                delete this;
            }

            byte           m_ch;

        private:
            details::_tcp_info *m_info;
            pplx::task_completion_event<int> m_op;
        };

        class _tcpstream_callback_agetc : public details::_filestream_callback
        {
        public:
            _tcpstream_callback_agetc(details::_tcp_info *info, pplx::task_completion_event<int> op) : m_op(op), m_info(info) { }

            virtual void on_completed(size_t result)
            {
                if ( result == sizeof(byte) )
                {
                    m_op.set(m_ch);   
                }
                else
                {
                    m_op.set(-1);
                }
                delete this;
            }

            byte           m_ch;

        private:
            details::_tcp_info *m_info;
            pplx::task_completion_event<int> m_op;
        };
#pragma endregion

        details::_tcp_info *m_info;
    };

    /// <summary>
    /// Stream buffer implementation for tcp sockets.
    /// </summary>
    class tcpstreambuf : public streambuf
    {
    public:
        /// <summary>
        /// Connect to a remote host and port or service.
        /// </summary>
        /// <param name="host">The name/address of the host</param>
        /// <param name="service">A port number or service name</param>
        /// <returns>An opened stream buffer.</returns>
        static pplx::task<streambuf> connect(const std::string &host, const std::string &service)
        {
            auto bfb = basic_tcpstreambuf::connect(host, service);
            return bfb.then([](pplx::task<basic_streambuf *> op) -> streambuf
                {
                    return streambuf(op.get());
                });
        }

        /// <summary>
        /// Connect to a remote host and port or service.
        /// </summary>
        /// <param name="host">The name/address of the host</param>
        /// <param name="port">A port number or service name</param>
        /// <returns>An opened stream buffer.</returns>
        static pplx::task<streambuf> connect(const std::string &host, int port)
        {
            std::stringstream ss;
            ss << port;

            auto bfb = basic_tcpstreambuf::connect(host, ss.str());
            return bfb.then([](pplx::task<basic_streambuf *> op) -> streambuf
                {
                    return streambuf(op.get());
                });
        }

        /// <summary>
        /// Connect to a remote host and port or service.
        /// </summary>
        /// <param name="host">The name/address of the host</param>
        /// <param name="service">A port number or service name</param>
        /// <returns>An opened stream buffer.</returns>
        static pplx::task<streambuf> connect(const std::wstring &host, const std::wstring &service)
        {
            auto bfb = basic_tcpstreambuf::connect(host, service);
            return bfb.then([](pplx::task<basic_streambuf *> op) -> streambuf
                {
                    return streambuf(op.get());
                });
        }

        /// <summary>
        /// Connect to a remote host and port or service.
        /// </summary>
        /// <param name="host">The name/address of the host</param>
        /// <param name="port">A port number or service name</param>
        /// <returns>An opened stream buffer.</returns>
        static pplx::task<streambuf> connect(const std::wstring &host, int port)
        {
            std::wstringstream ss;
            ss << port;

            auto bfb = basic_tcpstreambuf::connect(host, ss.str());
            return bfb.then([](pplx::task<basic_streambuf *> op) -> streambuf
                {
                    return streambuf(op.get());
                });
        }
    };
#endif // !defined(_cplusplus_winrt)
}

#pragma warning(pop) // 4100
